
/*
    信道模块设计
    服务器与客户端的通信是通过socket实现的，为了使资源得到更好的利用，引入了信道。
    一个socket连接可能会包含多个信道。
    发布者通过信道向服务器发布一条消息
    订阅者通过信道订阅指定队列的消息
*/
#ifndef __M_CHANNEL_H__
#define __M_CHANNEL_H__
#include "muduo/net/TcpConnection.h"
#include "muduo/prot/codec.h"
#include "muduo/prot/dispatcher.h"
#include "../mqcommon/helpper.hpp"
#include "../mqcommon/logger.hpp"
#include "../mqcommon/mq_message.pb.h" //消息的结构化描述在该部分使用protobuf生成
#include "../mqcommon/mq_prot.pb.h"
#include "../mqcommon/thread_pool.hpp"
#include "mq_consumer.hpp"
#include "mq_virtualhost.hpp"
#include "mq_route.hpp"

namespace xgmq
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using ThreadPoolPtr = std::shared_ptr<ThreadPool>;
    // 信道作为提供服务的接口，就会收到各种请求
    // 请求的时候，我们不需要接受太多的参数
    // 而是效仿muduo库，将收到的各种请求定义出来，定义出智能指针类型
    // 通过这些类型从而完成对应的操作
    // 所以需要首先定义出这些请求——这里使用protobuf
    // 上述过程也就是应用层协议的定制
    using openChannelRequestPtr = std::shared_ptr<openChannelRequest>;
    using closeChannelRequestPtr = std::shared_ptr<closeChannelRequest>;
    using declareExchangeRequestPtr = std::shared_ptr<declareExchangeRequest>;
    using deleteExchangeRequestPtr = std::shared_ptr<deleteExchangeRequest>;
    using declareQueueRequestPtr = std::shared_ptr<declareQueueRequest>;
    using deleteQueueRequestPtr = std::shared_ptr<deleteQueueRequest>;
    using queueBindRequestPtr = std::shared_ptr<queueBindRequest>;
    using queueUnBindRequestPtr = std::shared_ptr<queueUnBindRequest>;
    using basicPublishRequestPtr = std::shared_ptr<basicPublishRequest>;
    using basicAckRequestPtr = std::shared_ptr<basicAckRequest>;
    using basicConsumeRequestPtr = std::shared_ptr<basicConsumeRequest>;
    using basicCancelRequestPtr = std::shared_ptr<basicCancelRequest>;

    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;
        Channel(const std::string &cid,
                const VirtualHost::ptr &host,
                const muduo::net::TcpConnectionPtr &conn,
                const ProtobufCodecPtr &codec,
                const ConsumerManager::ptr &cmp,
                const ThreadPool::ptr &pool)
            : _cid(cid), _host(host), _conn(conn), _codec(codec), _cmp(cmp), _pool(pool) {
                DLOG("new channel:%p",this);
            }
        ~Channel()
        {
            // 当该信道要关闭时，要将该信道关联的消费者释放掉（避免内存泄漏）
            if (_consumer.get() != nullptr)
            {
                _cmp->remove(_consumer->qname, _consumer->tag);
            }
            DLOG("del channel:%p",this);
        }
        // 对外提供的服务
        // 交换机的声明与删除
        void declareExchange(const declareExchangeRequestPtr &req)
        {
            _host->declareExchange(req->exchange_name(), req->exchange_type(),
                                   req->durable(), req->auto_delet(),
                                   req->args());
            basicResopnse(true, req->cid(), req->rid());
        }
        void deleteExchange(const deleteExchangeRequestPtr &req)
        {
            _host->deleteExchange(req->exchange_name());
            basicResopnse(true, req->cid(), req->rid());
        }
        // 队列的声明与删除
        void declareQueue(const declareQueueRequestPtr &req)
        {
            bool ret = _host->declareQueue(req->queue_name(), req->durable(),
                                           req->exclusive(), req->auto_delet(),
                                           req->args());
            if (ret == false)
            {
                basicResopnse(false, req->cid(), req->rid());
                return;
            }
            // 初始化该队列的消费者管理句柄
            _cmp->initQueueConsumer(req->queue_name());
            basicResopnse(true, req->cid(), req->rid());
        }
        void deleteQueue(const deleteQueueRequestPtr &req)
        {
            // 删除队列的消费者
            _cmp->destoryQueueConsumer(req->queue_name());
            _host->deleteQueue(req->queue_name());
            basicResopnse(true, req->cid(), req->rid());
        }

        // 队列和交换机的 绑定与解绑
        void QueueBind(const queueBindRequestPtr &req)
        {
            _host->bind(req->exchange_name(), req->queue_name(), req->binding_key());
            basicResopnse(true, req->cid(), req->rid());
        }
        void QueueUnBind(const queueUnBindRequestPtr &req)
        {
            _host->unbind(req->exchange_name(), req->queue_name());
            basicResopnse(true, req->cid(), req->rid());
        }

        // 消息的发布
        void basicPublish(const basicPublishRequestPtr &req)
        {
            // 1,先获取交换机信息
            ExchangePtr exp = _host->getExchange(req->exchange_name());
            if (exp.get() == nullptr)
            {
                return basicResopnse(false, req->cid(), req->rid());
            }
            // 2，进行路由匹配，将消息发送给绑定的队列
            // 获取交换机的所有绑定信息
            MsgQueueBindingMap mqmp = _host->getExchangeBinding(req->exchange_name());
            std::string routing_key;
            basicProperties *bp = nullptr;
            if (req->has_properties())
            {
                bp = req->mutable_properties();
                routing_key = bp->routing_key();
            }
            for (auto &binding : mqmp)
            {
                if (Route::route(exp->type, binding.second->binding_key, routing_key))
                {
                    // 3，将消息添加到队列中（消息的管理）
                    _host->basePublish(binding.second->msgQueue_name, bp, req->body());
                    // 4，向线程池中添加消息的消费任务（从指定队列中获取一个消费者去消费）
                    auto task = std::bind(&Channel::consume, this, binding.first);
                    _pool->push(task);
                }
            }
            basicResopnse(true, req->cid(), req->rid());
        }
        // 消息的确认
        void basicAck(const basicAckRequestPtr &req)
        {
            // DLOG("信道模块，msg_id为：%s",req->msg_id().c_str());
            _host->ack(req->queue_name(), req->msg_id());
            basicResopnse(true, req->cid(), req->rid());
        }

        // 订阅队列的消息
        void basicConsume(const basicConsumeRequestPtr &req)
        {
            // 1，首先判断队列是否存在
            bool ret = _host->existsQueue(req->queue_name());
            if (ret == false)
            {
                return basicResopnse(false, req->cid(), req->rid());
            }
            // 2，创建消费者
            auto cb = std::bind(&Channel::callback, this,
                                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
            _consumer = _cmp->create(req->consume_tag(), req->queue_name(),
                                     req->auto_ack(), cb);
            basicResopnse(true, req->cid(), req->rid());
        }
        // 取消订阅
        void basicCancel(const basicCancelRequestPtr &req)
        {
            _cmp->remove(req->queue_name(), req->consum_tag());
            basicResopnse(true, req->cid(), req->rid());
        }

    private:
        // 消费者内部的回调函数
        // 将消息推送给指定的消费者
        void callback(const std::string &tag, const basicProperties *bp, const std::string &body)
        {
            // DLOG("调用消费者回调函数，将消息发给消费者\n");
            basicConsumeResponse resp;
            resp.set_cid(_cid);
            resp.set_consum_tag(tag);
            resp.set_body(body);
            if (bp)
            {
                resp.mutable_properties()->set_id(bp->id());
                resp.mutable_properties()->set_routing_key(bp->routing_key());
                resp.mutable_properties()->set_deliver_mode(bp->deliver_mode());
            }
            _codec->send(_conn, resp);
        }
        void consume(const std::string &qname)
        {
            // 1,取出队列中的一条消息
            MessagePtr mp = _host->basicConsume(qname);
            if (mp.get() == nullptr)
            {
                DLOG("消费消息时，对应队列的消息不存在\n");
                return;
            }
            // 2，取出一个消费者
            Consumer::ptr cp = _cmp->choose(qname);
            if (cp.get() == nullptr)
            {
                DLOG("执行任务失败,队列没有 消费者\n");
                return;
            }
            // 3，调用消费者对应的消息处理回调函数
            cp->callback(cp->tag, mp->mutable_payload()->mutable_properties(), mp->payload().body());
            // 4，判断该消费者是否自动确认
            if (cp->auto_delete)
            {
                _host->ack(qname, mp->payload().properties().id());
            }
        }
        void basicResopnse(bool ok, const std::string &cid, const std::string &rid)
        {
            basicCommonResponse resp;
            resp.set_ok(ok);
            resp.set_cid(cid);
            resp.set_rid(rid);
            _codec->send(_conn, resp);
        }

    private:
        std::string _cid;                   // 信道id，信道的唯一标识
        Consumer::ptr _consumer;            // 该信道所关联的消费者
        VirtualHost::ptr _host;             // 虚拟机句柄，包含了交换机，队列，绑定关系，消息等模块
        muduo::net::TcpConnectionPtr _conn; // 连接管理
        ProtobufCodecPtr _codec;            // 协议处理句柄
        ConsumerManager::ptr _cmp;          // 消费者管理句柄
        ThreadPool::ptr _pool;                // 线程池
    };

    // 信道的管理
    // 完成信道增删查
    class ChannelManager
    {
    public:
        using ptr=std::shared_ptr<ChannelManager>;
        ChannelManager() {}
        bool openChannel(const std::string &cid,
                         const VirtualHost::ptr &host,
                         const muduo::net::TcpConnectionPtr &conn,
                         const ProtobufCodecPtr &codec,
                         const ConsumerManager::ptr &cmp,
                         const ThreadPool::ptr &pool)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(cid);
            if (it != _channels.end())
            {
                return false;
            }
            auto channel = std::make_shared<Channel>(cid, host, conn, codec, cmp, pool);
            _channels.insert(std::make_pair(cid, channel));
            return true;
        }
        void closeChannel(const std::string &cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it=_channels.find(cid);
            if(it==_channels.end())
            {
                return;
            }
            _channels.erase(it);
        }
        Channel::ptr selectChannel(const std::string &cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(cid);
            if (it != _channels.end())
            {
                return it->second;
            }
            return Channel::ptr();
        }

    private:
        std::mutex _mutex;
        std::unordered_map<std::string, Channel::ptr> _channels;
    };

}
#endif